package concurrent.threadscheduling;

import concurrent.test.FutureTaskTest;
import util.GsonUtil;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 多个线程并发的优化.
 *
 * 举栗：
 * 有一个接口内部，调用了10个方法，这10个方法互不影响，应该如何优化？
 *
 * 答：
 * 1、并行化执行。以耗时最久的那个接口作为接口耗时。
 * 2、如果某个方法执行失败，则快速失败，通知其他操作中断，要求异步的方法需要响应中断。
 * 3、增加超时机制，最大等待不超过X秒，并且知道是哪个操作失败or超时了，作为状态码返回。
 * 4、如果方法执行超时，或者有一个方法失败了，响应中断，并且其他任务快速失败，然后执行rollback。
 *
 * 实现
 * 1、使用 CountDownLatch, 在异步线程里执行的任务，每执行完成后，countDown()一次，并且向结果集中写入数据
 * 2、使用AtomicBoolean 作为中断标记位，当超时或者有方法执行失败时，设置中断标记位为true。异步方法内部，设置中断检查点，检查并响应中断
 *
 * Created by yzy on 2021-02-03 15:15
 */
public class MultiThreadConcurrent {
    private static final ExecutorService EXECUTORS = Executors.newCachedThreadPool();


    public static void main(String[] args) {
        long start = System.currentTimeMillis();
        testMethod(6, TimeUnit.SECONDS);
        long end = System.currentTimeMillis();
        System.out.println("执行时间:" + (end- start));
    }


    private static void testMethod(long timeout, TimeUnit timeUnit){
        AtomicBoolean interrupt = new AtomicBoolean(false);
        final CountDownLatch cd = new CountDownLatch(10);
        List<String> resList = new CopyOnWriteArrayList<>();
        Future<?> future1 = EXECUTORS.submit(() -> {
            String res = null;
            try {
                res = processMethod("Thread_1", 10, "111", interrupt);
                resList.add(res);
                cd.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Future<?> future2 = EXECUTORS.submit(() -> {
            String res = null;
            try {
                res = processMethod("Thread_2",5, null, interrupt);
                resList.add(res);
                cd.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Future<?> future3 = EXECUTORS.submit(() -> {
            String res = null;
            try {
                res = processMethod("Thread_3",8, "333", interrupt);
                resList.add(res);
                cd.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Future<?> future4 = EXECUTORS.submit(() -> {
            String res = null;
            try {
                res = processMethod("Thread_4",7, "444", interrupt);
                resList.add(res);
                cd.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Future<?> future5 = EXECUTORS.submit(() -> {
            String res = null;
            try {
                res = processMethod("Thread_5",6, "555", interrupt);
                resList.add(res);
                cd.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Future<?> future6 = EXECUTORS.submit(() -> {
            String res = null;
            try {
                res = processMethod("Thread_6",4, "666", interrupt);
                resList.add(res);
                cd.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Future<?> future7 = EXECUTORS.submit(() -> {
            String res = null;
            try {
                res = processMethod("Thread_7",2, "777", interrupt);
                resList.add(res);
                cd.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Future<?> future8 = EXECUTORS.submit(() -> {
            String res = null;
            try {
                res = processMethod("Thread_8",3, "888", interrupt);
                resList.add(res);
                cd.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Future<?> future9 = EXECUTORS.submit(() -> {
            String res = null;
            try {
                res = processMethod("Thread_9",3, "999", interrupt);
                resList.add(res);
                cd.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Future<?> future10 = EXECUTORS.submit(() -> {
            String res = null;
            try {
                res = processMethod("Thread_10",8, "AAA", interrupt);
                resList.add(res);
                cd.countDown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        try {
            cd.await(timeout, timeUnit);
            if(cd.getCount() != 0){
                System.out.println("count:" + cd.getCount());
                interrupt.set(true);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("执行完成:" + GsonUtil.toJson(resList));

        if(interrupt.get() == true){
            System.out.println("异步方法被打断, 这里执行回滚方法...");
            return;
        }
    }


    private static String processMethod(String name, int second, String val, AtomicBoolean interrupt) throws InterruptedException {
        for(int i=0; i<second; i++){
            if(interrupt.get() == true){
                throw new InterruptedException();
            }
            Thread.sleep(1000);
        }
        if(val == null){
            System.out.println("当前线程执行失败. name=" + name);
            interrupt.set(true);
        }
        System.out.println("Current method " + name + " costs: " + second + ", val=" + val);
        return val;
    }

}
